package com.client.rpcclient.impl;



import com.common.loadbalance.impl.CircleLoadBalance;
import com.client.rpcclient.NettyClientInitializer;
import com.client.rpcclient.RPCClient;
import com.client.util.SingletonFactory;
import com.client.util.UnProcessRequests;
import com.client.zk.ZkService;
import com.common.entity.RPCRequest;
import com.common.entity.RPCResponse;
import com.common.serialize.CommonSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

/**
 * @author Zixiang Hu
 * @description 使用netty实现RPC客户端
 * @create 2021-03-01-15:04
 */
@Slf4j
@AllArgsConstructor
@Builder
public class NettyRPCClient implements RPCClient {

    private CommonSerializer serializer;
    private static final EventLoopGroup group;
    private static final Bootstrap bootstrap;
    private static UnProcessRequests unProcessRequests;

    private  ZkService zkService;

    static {
        // 这里根据单例工厂获取
        unProcessRequests = SingletonFactory.getObject(UnProcessRequests.class);
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class);
    }


    @Override
    public CompletableFuture<RPCResponse> sendRequest(RPCRequest rpcRequest) {
        try {
            InetSocketAddress address = zkService.discoveryService(rpcRequest.getInterfaceName());
            if (address == null)
                return null;
            CompletableFuture<RPCResponse> future = new CompletableFuture<>();
            unProcessRequests.put(rpcRequest.getRequestId(), future);
            bootstrap.handler(new NettyClientInitializer(serializer));
            ChannelFuture channelFuture = bootstrap.connect(address.getHostString(), address.getPort()).sync();
            Channel channel = channelFuture.channel();
            channel.writeAndFlush(rpcRequest);
            return future;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
        }
        return null;
    }
}
